package com.sduept.component.message.core;

import com.sduept.nwld.dataserver.manager.etl.*;
import lombok.extern.slf4j.Slf4j;
import net.sf.json.JSONObject;
import org.apache.commons.lang.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

import java.util.Optional;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@Slf4j
@Service
public class ProtectDeviceMessageService {
    @Autowired
    private ProtectChannelAlarmETLManager manager;
    @Autowired
    private ProtectEquipmentETLManager proEManager;
    @Autowired
    private ProtectDeviceAlarmETLManager deviceManager;
    @Autowired
    private ProtectionDeviceActionETLManager actionManager;
    @Autowired
    private DeviceFaultLocationETLManager faultManager;
    private ExecutorService executor = Executors.newFixedThreadPool(10);

    @KafkaListener(topics = {"protection_channel_alarm"})
    public void onMessage(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            JSONObject jo = JSONObject.fromObject(kafkaMessage.get());
            String msgContent = jo.getString("content");
            JSONObject msgJson = JSONObject.fromObject(msgContent);
            final String datetime = msgJson.getString("updatetime");
            log.info("接收到消息：protection_channel_alarm, 数据表：t_protection_channel_alarm,数据更新时间："+datetime);
            if (!StringUtils.isEmpty(datetime)) {
                executor.submit(new Runnable() {
                    @Override
                    public void run() {
                        manager.receiveAndSavePartData(datetime);
                    }
                });
            }
        }
    }

    @KafkaListener(topics = {"cim_protection_equipment_update"})
    public void onMessage1(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            JSONObject jo = JSONObject.fromObject(kafkaMessage.get());
            String msgContent = jo.getString("content");
            JSONObject msgJson = JSONObject.fromObject(msgContent);
            final String datetime = msgJson.getString("updatetime");
            log.info("接收到消息：cim_protection_equipment_update, 数据表：t_bx_protection_equipment,数据更新时间："+datetime);
            if (!StringUtils.isEmpty(datetime)) {
                executor.submit(new Runnable() {
                    @Override
                    public void run() {
                        proEManager.incrementalUpdating(datetime);
                    }
                });
            }
        }
    }

    @KafkaListener(topics = {"protection_device_alarm"})
    public void onMessage2(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            JSONObject jo = JSONObject.fromObject(kafkaMessage.get());
            String msgContent = jo.getString("content");
            JSONObject msgJson = JSONObject.fromObject(msgContent);
            final String datetime = msgJson.getString("updatetime");
            log.info("接收到消息：protection_device_alarm, 数据表：t_bx_protect_device_alarm,数据更新时间："+datetime);
            if (!StringUtils.isEmpty(datetime)) {
                executor.submit(new Runnable() {
                    @Override
                    public void run() {
                        deviceManager.receiveAndSavePartData(datetime);
                    }
                });
            }
        }
    }

    @KafkaListener(topics = {"protection_action_data"})
    public void onMessage3(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        JSONObject jo = JSONObject.fromObject(kafkaMessage.get());
        String msgContent = jo.getString("content");
        JSONObject msgJson = JSONObject.fromObject(msgContent);
        final String datetime = msgJson.getString("updatetime");
        log.info("接收到消息：protection_action_data, 数据表：t_bx_protect_device_action,数据更新时间："+datetime);
        if (!StringUtils.isEmpty(datetime)) {
            executor.submit(new Runnable() {
                @Override
                public void run() {
                    actionManager.receiveAndSavePartData(datetime);
                }
            });
        }
    }


    @KafkaListener(topics = {"device_fault_location"})
    public void onMessage4(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            JSONObject jo = JSONObject.fromObject(kafkaMessage.get());
            String msgContent = jo.getString("content");
            JSONObject msgJson = JSONObject.fromObject(msgContent);
            final String datetime = msgJson.getString("updatetime");
            log.info("接收到消息：device_fault_location, 数据表：t_bx_device_fault_location,数据更新时间："+datetime);
            if (!StringUtils.isEmpty(datetime)) {
                executor.submit(new Runnable() {
                    @Override
                    public void run() {
                        faultManager.receiveAndSavePartData(datetime);
                    }
                });
            }
        }

    }
}
